package com.bigdata.sql

import org.apache.spark.sql.{DataFrame, SparkSession}

object Demo2DSLWC {


  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession
      .builder()
      .master("local")
      .appName("sql")
      //指定spark sql  在shuffle之后分区数，默认时200， 类似hive中设计reduce的数量
      .config("spark.sql.shuffle.partitions", 1)
      .getOrCreate()

    spark.sparkContext.setLogLevel("error")

    val linesDF: DataFrame = spark
      .read
      .format("csv")
      .schema("lines string")
      .option("sep", "\t")
      .load("data/words.txt")

    /**
     * DSL： 类sql api ,介于代码和sql之间的一种写法
     */
    // 导入spark sql所有的函数，才能在代码中使用函数
    import org.apache.spark.sql.functions._
    // 导入spark sql的隐式转换，才能使用$符号引用字段
    import spark.implicits._

    //    val studentDF = spark.read.format("csv").option("sep", ",")
    //      .schema("id STRING,name STRING,age INT,gender STRING,clazz STRING")
    //      .load("/data/student")
    //
    //    studentDF.createOrReplaceTempView("t_stu")
    //
    //    val result = spark.sql(
    //      """
    //        |select gender, count(gender) as c from t_stu group by gender
    //        |""".stripMargin)
    //    result.show()

    // DSL 没有子查询的说法，上面查询到的字段，下面的方法可以直接使用
    linesDF
      .select(explode(split($"lines", ",")) as "word")
      .groupBy($"word")
      .agg(count($"word") as "dsl_count")
      .select($"dsl_count")
      .show(100, false)


  }

}
